package com.fanxuankai.boot.mqbroker.consumer.autoconfigure;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.fanxuankai.boot.mqbroker.Event;
import com.fanxuankai.boot.mqbroker.autoconfigure.MqBrokerProperties;
import com.fanxuankai.boot.mqbroker.consumer.AbstractMqConsumer;
import com.fanxuankai.boot.mqbroker.consumer.EventListenerRegistry;
import com.fanxuankai.boot.mqbroker.consumer.MqConsumer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.MessageListener;

import java.util.Optional;

/**
 * @author fanxuankai
 */
@ConditionalOnClass({KafkaTemplate.class})
@ConditionalOnProperty(name = MqBrokerConsumerProperties.ENABLED, matchIfMissing = true)
public class MqBrokerConsumerKafkaAutoConfiguration {
    @Bean
    public AbstractMessageListenerContainer<String, String> messageListenerContainer(
            ConsumerFactory<String, Object> consumerFactory, AbstractMqConsumer<String> mqConsumer,
            EventListenerRegistry eventListenerRegistry, KafkaProperties kafkaProperties) {
        String[] topics = eventListenerRegistry.getTopics().toArray(new String[0]);
        ContainerProperties containerProperties = new ContainerProperties(topics);
        containerProperties.setMessageListener((MessageListener<String, String>) data -> mqConsumer.accept(data.value()));
        ConcurrentMessageListenerContainer<String, String> concurrentMessageListenerContainer =
                new ConcurrentMessageListenerContainer<>(consumerFactory, containerProperties);
        Optional.ofNullable(kafkaProperties.getListener().getConcurrency())
                .ifPresent(concurrentMessageListenerContainer::setConcurrency);
        containerProperties.getKafkaConsumerProperties().putAll(kafkaProperties.getConsumer().getProperties());
        return concurrentMessageListenerContainer;
    }

    @Bean
    @ConditionalOnMissingBean(MqConsumer.class)
    public AbstractMqConsumer<String> mqConsumer(MqBrokerProperties properties) {
        return new AbstractMqConsumer<String>(properties) {
            @Override
            public Event<String> apply(String s) {
                return JSON.parseObject(s, new TypeReference<Event<String>>() {
                });
            }
        };
    }
}
